package snippets.cse524.activeDates;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import snippets.utils.IntervalEvaluator;
import snippets.utils.Utils;

public class AllIntervalEvaluator extends Configured implements Tool {

	public static class Map extends MapReduceBase implements
			Mapper<LongWritable, Text, Text, Text> {
		private static final IntWritable ONE = new IntWritable(1);
		public int[] GAPS = { 1, 5, 10, 15, 20 };
		private String methodPrefix = null;
		private static final java.util.Map<String, IntervalEvaluator> evaluators = new HashMap<String, IntervalEvaluator>();

		public void configure(JobConf job) {
			// TODO Auto-generated method stub
			super.configure(job);
			methodPrefix = job.get("methodPrefix");
			assert (methodPrefix != null);
		}

		static {
			for (double i = 0.4; i <= 1; i += 0.1) {
				evaluators.put("MIN-PERIOD-FOR-RATIO:" + i,
						Utils.getThresholdedIntervalEvaluator(i));
			}

			for (double i = 0; i <= 1; i+=0.1) {
				evaluators.put("MOST_WEIGHTED-PERIOD-BY-FRACTION " + i,
						Utils.getFractionInternalEvaluator(i));
			}
		}

		@Override
		public void map(LongWritable key, Text line,
				OutputCollector<Text, Text> collector, Reporter arg3)
				throws IOException {

			String content = line.toString();

			String[] parts = StringUtils.split(content, '\t');

			String pageId = parts[0];
			double mentions = Double.parseDouble(parts[1]);

			java.util.Map<Integer, Double> yearCounts = new HashMap<Integer, Double>();

			for (String part : ArrayUtils.subarray(parts, 2, parts.length)) {
				String[] ycParts = StringUtils.split(part, ':');

				yearCounts.put(Integer.parseInt(ycParts[0]),
						Double.parseDouble(ycParts[1]));
			}
			
			if(yearCounts.isEmpty()) return;
			List<String> evaluatorsStr = new ArrayList<String>(evaluators.keySet());
			Collections.sort(evaluatorsStr);
			
			List<String> opParts = new ArrayList<String>();
			
			for(String method : evaluatorsStr) {
				
				// TODO: Use heuristics here.
				int left = Collections.min(yearCounts.keySet()) + 35;
				int right = Collections.min(yearCounts.keySet()) + 45;
				
				Pair<Integer, Integer> period = evaluators.get(method).getInterval(yearCounts);
				if(period != null) {
					left = period.getLeft();
					right = period.getRight();
				}
				
				opParts.add(methodPrefix + method);
				opParts.add(Integer.toString(left));
				opParts.add(Integer.toString(right));
			}
			
			String normalizedTitle = StringUtils.replaceChars(pageId, ' ', '_');
			
			collector.collect(new Text(normalizedTitle), new Text(StringUtils.join(opParts, '\t')));
		}
	}

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		ToolRunner.run(new AllIntervalEvaluator(), args);

	}

	@Override
	public int run(String[] args) throws Exception {
		String inputPath = args[0];
		String outputPath = args[1];

		JobConf conf = new JobConf(getConf(), AllIntervalEvaluator.class);
		conf.setJobName(String.format("AllInternalEvaluator [%s: %s]",
				inputPath, outputPath));

		conf.set("methodPrefix", args[2]);

		conf.setNumMapTasks(12);
		conf.setNumReduceTasks(12);

		FileInputFormat.setInputPaths(conf, new Path(inputPath));
		FileOutputFormat.setOutputPath(conf, new Path(outputPath));

		conf.setInputFormat(TextInputFormat.class);
		conf.setOutputFormat(TextOutputFormat.class);

		conf.setMapperClass(Map.class);

		// Set the outputs for the Map
		// PairOfWritables<ArrayListWritable<PairOfStrings>,
		// ArrayListWritable<PairOfWritables<Text,
		// ArrayListWritable<PairOfWritables<DoubleWritable,
		// PairOfWritables<IntWritable, PairOfInts>>>>>> mapOpDummy = new
		// PairOfWritables<ArrayListWritable<PairOfStrings>,
		// ArrayListWritable<PairOfWritables<Text,
		// ArrayListWritable<PairOfWritables<DoubleWritable,
		// PairOfWritables<IntWritable, PairOfInts>>>>>>();
		conf.setMapOutputKeyClass(Text.class);
		conf.setMapOutputValueClass(Text.class);

		JobClient.runJob(conf);

		return 0;
	}
}
